/*
    Copyright 2010 Northbranchlogic, Inc.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.norbl.runjrun;

import com.xerox.amazonws.sdb.*;
import com.norbl.util.*;
import com.norbl.util.sdb.*;
import com.norbl.util.s3.*;
import com.norbl.util.sqs.*;
import com.xerox.amazonws.sqs2.SQSException;
import java.util.*;

/** Access to the SDB 'table' of {@link TaskState}s,
 *  where each element one 
 *
 * @author Barnet Wagman
 */
public class TaskStateTable extends DomainWrangler {

    public TaskState[] currentStates;
    NodeTaskOps ops;

    public TaskStateTable(SdbOps dops) throws SDBException {
        super(dops,Constants.TASK_TABLE_NAME);
    }

    public void setOps(NodeTaskOps ops) {
        this.ops = ops;
    }

    public TaskStateTable cleanClone(SdbOps newOps) {
        try {
            return(new TaskStateTable(newOps) );
        }
        catch(SDBException x) { throw new RuntimeException(x); }
    }

    public DbEl toDbEl(PropertyList pl,boolean allowMissingFields) {
        return( new TaskState(pl,allowMissingFields) );
    }

    public TaskState[] getTaskStates() throws SDBException {
        List<DbEl> els = getItems(true);
        return( els.toArray(new TaskState [els.size()]) );
    }

    public DbEl getTemplateDbEl() { return(new TaskState()); }

    public TaskState getTaskState(String taskName) {
        if ( taskName == null ) return(null);
        if ( currentStates == null ) return(null);
        for ( TaskState s : currentStates ) {
            if ( taskName.equals(s.taskName) ) return(s);
        }
        return(null);
    }

    public void setState(String taskName,TaskState taskState,boolean retry)
        throws SDBException {
        putItem(taskName,taskState,retry);
    }

    public void recordNewTask(TaskDescription taskDes,
                              long submissionTime)
        throws SDBException {

        setState(taskDes.inputS3Key,
                 new TaskState(taskDes.inputS3Key,
                              taskDes,
                              TaskState.Status.waiting,
                              null,// host
                              null, // instance ID
                              new Long(submissionTime), // status time
                              new Long(submissionTime), // submission time
                              null, // start time
                              null // finish time
                            ),
                 true);
    }

    public void setTaskStateRunning(String host,String instanceID,
                                    TaskDescription des,
                                    long tm) throws SDBException {
       
        setState(des.inputS3Key,
            new TaskState(des.inputS3Key,
                          des,
                          TaskState.Status.running,
                          host,
                          instanceID,
                          new Long(tm), // status time
                          null, // submission time
                          new Long(tm), // start time
                          null // finished time
                          ),
            true);
    }

     void setTaskStateFinished(String host, String instanceID,
                               TaskDescription des,
                               long tm) throws SDBException {

        setState(des.inputS3Key,
            new TaskState(des.inputS3Key,
                          des,
                          TaskState.Status.finished,
                          host,
                          instanceID,
                          new Long(tm), // status time
                          null, // submission time
                          null, // start time
                          new Long(tm) // finished time
                          ),
            true);
    }

    void setTaskStateTerminatedByException(String host, String instanceID,
                                           TaskDescription des,
                                           long tm)
        throws SDBException {

        setState(des.inputS3Key,
            new TaskState(des.inputS3Key,
                          des,
                          TaskState.Status.terminatedByException,
                          host,
                          instanceID,
                          new Long(tm), // status time
                          null, // submission time
                          null, // start time
                          new Long(tm) // finished time
                          ),
            true);
    }

    void setTaskStateMissingRunnable(String host, String instanceID,
                                     TaskDescription des, long tm)
        throws SDBException  {

        setState(des.inputS3Key,
            new TaskState(des.inputS3Key,
                          des,
                          TaskState.Status.missingRunnable,
                          host,
                          instanceID,
                          new Long(tm), // status time
                          null, // submission time
                          null, // start time
                          new Long(tm) // finished time
                          ),
          true);

    }

    public void recordRequeuedTask(TaskDescription taskDes,
                                   long tm) throws SDBException {

        setState(taskDes.inputS3Key,
                 new TaskState(taskDes.inputS3Key,
                              taskDes,
                              TaskState.Status.waiting,
                              null,// host
                              null, // instance ID
                              new Long(tm), // status time
                              new Long(tm), // submission time
                              null, // start time
                              null // finish time
                            ),
                 true);
    }

    public void recordOrphanedTask(String taskName,
                                   long tm) throws SDBException {

        setState(taskName,
                 new TaskState(taskName,
                              null, // task des
                              TaskState.Status.orphan,
                              Constants.NIL,// host
                              Constants.NIL, // instance ID
                              new Long(tm), // status time
                              null, // submission time
                              null, // start time
                              null // finish time
                            ),
                 true);
    }

    public void recordUnrunnableTask(String taskName,
                                   long tm) throws SDBException {

        setState(taskName,
                 new TaskState(taskName,
                              null, // task des
                              TaskState.Status.unrunnable,
                              Constants.NIL,// host
                              Constants.NIL, // instance ID
                              new Long(tm), // status time
                              null, // submission time
                              null, // start time
                              null // finish time
                            ),
                 true);
    }

        /** Gets an updated copy of the task states (from sdb).<p>
         *
         *  This method also checks for orphaned tasks - tasks that are
         *  listed as running on nodes that are not running or no longer exist.
         *  A list of orphans are returned.
         *         
          * @param nodeStates this must be an up to date list of nodes.
          * @param sops
          * @param qops
          * @return a list of all orphans found or an empty list.
          * @throws SDBException
          */
    public List<TaskState> update(NodeState[] nodeStates,S3Ops sops, SqsOps qops)
        throws SDBException {
       
        currentStates = getTaskStates();
       
        List<TaskState> orphans = new ArrayList<TaskState>();
        long tm = System.currentTimeMillis();

            // Check for orphans
        NEXT: for ( TaskState t : currentStates ) {    
            
                // If the task not marked running or is already marked
                // as an orphan, there's nothing to do here.
            if ( !t.status.equals(TaskState.Status.running) ||
                 t.status.equals(TaskState.Status.orphan) ||
                 t.status.equals(TaskState.Status.unrunnable) ) {
                continue NEXT;
            }
            
            NodeState node = NodeStateTable.getNode(nodeStates,t.host);
            if ( (node == null) || !node.isRunning() ) { // An orphan
                orphans.add(t);
            }
        }
        return(orphans);
    }

    public void requeueOrphans(List<TaskState> orphans) 
        throws SDBException, SQSException {
        for ( TaskState o : orphans ) {
            ops.requeueTask(o);
        }
    }

    public void recordOrphanedTasks(List<TaskState> orphans)
        throws SDBException {
        long tm = System.currentTimeMillis();
        for ( TaskState o : orphans ) {
            recordOrphanedTask(domainName, tm);
        }
    }
}
